"""
FiftyOne Server utils.

| Copyright 2017-2025, Voxel51, Inc.
| `voxel51.com <https://voxel51.com/>`_
|
"""

import typing as t

import cachetools
from dacite import Config, from_dict as _from_dict
from dacite.core import T
from dacite.data import Data

import fiftyone.core.dataset as fod
from fiftyone.core.expressions import ViewField as F
import fiftyone.core.fields as fof
import fiftyone.core.media as fom
from fiftyone.server.utils import http, json


_cache = cachetools.TTLCache(maxsize=10, ttl=900)  # ttl in seconds
_dacite_config = Config(check_types=False)


def load_and_cache_dataset(name):
    """Loads the dataset with the given name and caches it.

    This method is a wrapper around :func:`fiftyone.core.dataset.load_dataset`
    that stores a reference to every dataset it loads in a TTL cache so that
    subsequent calls to this method can return the cached dataset object rather
    than reloading it. This is useful in situations where expensive objects
    (eg brain results) are cached on the dataset and we want to avoid reloading
    them.

    Args:
        name: the dataset name

    Returns:
        a :class:`fiftyone.core.dataset.Dataset`
    """
    return _cache_dataset(name)


def cache_dataset(dataset):
    """Returns a cached version of the given dataset.

    If this dataset has already been cached, then the cached copy will be
    returned. Otherwise, the provided dataset will be inserted into the cache
    and then returned.

    See :func:`load_and_cache_dataset` for more context.

    Args:
        dataset: a :class:`fiftyone.core.dataset.Dataset`

    Returns:
        a :class:`fiftyone.core.dataset.Dataset`
    """
    return _cache_dataset(dataset.name, dataset=dataset)


def _cache_dataset(name, dataset=None):
    _dataset = _cache.get(name)

    if _dataset is not None:
        try:
            _dataset.reload()
            return _dataset
        except:
            # dataset was deleted
            pass

    if dataset is not None:
        _dataset = dataset
    else:
        _dataset = fod.load_dataset(name, reload=True)

    _cache[name] = _dataset

    return _dataset


def change_sample_tags(sample_collection, changes):
    """Applies the changes to tags to all samples of the collection, if
    necessary.

    Args:
        sample_collection: a
            :class:`fiftyone.core.collections.SampleCollection`
        changes: a dict of tags as keys and bools as values. A ``True`` value
            adds the tag to all samples, if necessary. A ``False`` value
            removes the tag from all samples, if necessary
    """
    add_tags, del_tags = _parse_changes(changes)

    if add_tags:
        sample_collection.tag_samples(add_tags)

    if del_tags:
        sample_collection.untag_samples(del_tags)


def change_label_tags(sample_collection, changes, label_fields=None):
    """Applies the changes to tags to all labels in the specified label
    field(s) of the collection, if necessary.

    Args:
        sample_collection: a
            :class:`fiftyone.core.collections.SampleCollection`
        changes: a dict of tags as keys and bools as values. A ``True`` value
            adds the tag to all labels, if necessary. A ``False`` value removes
            the tag from all labels, if necessary
        label_fields (None): an optional name or iterable of names of
            :class:`fiftyone.core.labels.Label` fields. By default, all label
            fields are used
    """
    add_tags, del_tags = _parse_changes(changes)

    if add_tags:
        sample_collection.tag_labels(add_tags, label_fields=label_fields)

    if del_tags:
        sample_collection.untag_labels(del_tags, label_fields=label_fields)


def from_dict(data_class: t.Type[T], data: Data) -> T:
    """Wrapping function for ``dacite.from_dict`` that ensures a common
    configuration is used.

    Args:
        data_class: a dataclass
        data: the data with which to instantiate the dataclass instance

    Returns:
        a dataclass instance
    """
    return _from_dict(data_class, data, config=_dacite_config)


def meets_type(field: fof.Field, type_or_types):
    """
    Determines whether the field meets type or types, or the field
    is a :class:`fiftyone.core.fields.ListField` that meets the type or types

    Args:
        field: a class:`fiftyone.core.fields.Field`
        type: a field type or `tuple` of field types
    """
    return isinstance(field, type_or_types) or (
        isinstance(field, fof.ListField)
        and isinstance(field.field, type_or_types)
    )


def attach_frame_if_necessary(view, frame_number, modal, group_slice=None):
    """Attaches a frame to view when applicable

    Args:
        view: the dataset view
        frame_number: a frame number
        modal: whether this is an App modal request
        group_slice (None): the group slice

    Returns:
        a new view and whether the attachment is true
    """
    is_video = view.media_type == fom.VIDEO
    if view.media_type == fom.MIXED:
        is_video = view._dataset.group_media_types[group_slice] == fom.VIDEO

    if is_video and frame_number is not None:
        default_filter = F("frame_number") == 1
        current_filter = F("frame_number").is_in([frame_number, 1])
        filter_frames = lambda f: F("frames").filter(f)
        expr = F.if_else(
            F("_id").to_string() == modal,
            filter_frames(current_filter),
            filter_frames(default_filter),
        )
        view = view.set_field("frames", expr)

    return view, is_video


def _parse_changes(changes):
    add_tags = []
    del_tags = []
    for tag, add in changes.items():
        if add:
            add_tags.append(tag)
        else:
            del_tags.append(tag)

    return add_tags, del_tags
